package spark_core

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.window
import org.apache.spark.{SparkConf, SparkContext}

import java.sql.Timestamp
import java.util.Properties


object Test {
  def main(args: Array[String]): Unit = {

    println("hello spark")
//    //配置环境上下文,在服务器中也可配置
    val conf=new SparkConf().setAppName("wordContDemo")
    val sc=new SparkContext(conf)
    //读取指定地方的文件，读取出来是一个RDD的文件，之后在使用RDD算子，不常用
    /**
     * flatMap()算子里面放着一个处理逻辑或函数
     * split()做切割(分割)
     * map()编程<k,v>键值模式
     * reduceByKey()统计键值对中的和
     * saveAsTestFile()作为文件存储
     * persist()数据持久化（用到数据重用），unpersist()去除持久化
     * 分区：哈希分区和区域分区
     * join内连接：只有两个集中都存在才输出
     *
     * spark中的JSON格式数据是指一行
     * Dataframes中的对象是Row的对象，不知道列和字段更通用，缺少类型检查（效率更好）
     * Dataset知道字段，字段类型，可以做编译器的检查
     * SparkSession
     *
     * DataFrame查看数据：传统数据API和sparkAPI，
     *     处理时将DataFrame注册为一张临时表，然后通过sql语句进行查询
     *     处理时将DataFrame注册为一张全局表，然后通过sql语句进行查询
     *
     *     limit获取DataFrame的前n行，不是action
     *     groupBy根据字段进行分组操作
     *
     *
     *
     * */
//    sc.textFile(args(0)).flatMap(line=>line.split(" "))
//      .map(word=>(word,1))
//      .reduceByKey((a,b)=>a+b)
//      .saveAsTextFile(args(1))

    val spark=SparkSession.builder().appName("dataFrameDemo").master("local[*]").getOrCreate()
    import spark.implicits._
    //读取excel文件

//    val csvDF=spark.read.format("csv").option("header","true")
//      .option("mode","FAILFAST")
//      .load(args(0))
//
//    csvDF.show(3)
//    csvDF.printSchema()

//    csvDF.filter("DEST_COUNTRY_NAME='United States'")
//    csvDF.createOrReplaceTempView("fly")
//    spark.sql("select * from fly where count>10")
//    spark.sql("select * from fly where count>900").show(5)
//    spark.sql("select * from fly where count>900").write
//      .csv(args(0))

//    读取mysql数据库
    val url="jdbc:mysql://localhost:3306/test?useUnicode=true val table=\"user\"\n//    val properties=new Properties()\n//    properties.setProperty(\"user\",\"root\")\n//    properties.setProperty(\"password\",\"123456789\")&characterEncoding=utf-8&useSSL=false"
//
    //jdbc分区作用
//    val sqlDF=spark.read.jdbc(url, table,"userid",1000L,2000L,6, properties)
//    val sqlDF=spark.read.jdbc(url, table,"id",10000L,2000000L,5, properties)
//   println(sqlDF.rdd.getNumPartitions)

//    sqlDF.select("username","usersex").distinct().show(5)
//    println(sqlDF.rdd.getNumPartitions)
//    sqlDF.createOrReplaceTempView("book")
//     spark.sql("select * from book limit 10").show()
//     spark.sql("select * from book limit 10").write.csv(args(0))

    //sparkStreaming实例

    //准备数据源
    val lines=spark.readStream.format("socket")
      .option("host","192.168.80.133")
      .option("port","9999")
      .option("includeTimestamp",true)
      .load()
    val wordCount = lines.as[(String,Timestamp)]
      .flatMap(lines => lines._1.split(" ").map(word=>(word,lines._2)))
      .toDF("word","timestamp")
      .groupBy(window($"timestamp","10 minutes","5 minutes"),$"word")
      .count()
    //输出结果集
    val writer=new JdbcUtil(url,"root","123456789")
    val query=wordCount.writeStream
      //自定义数据输出格式
      .foreach(writer)
      .format("console")
      .outputMode("complete")
      .outputMode("update")
      .start()
    query.awaitTermination()
//      nc -l  192.168.60.130 9999


  }

}
